package com.bw.yk08;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;

public class HbaseSinkFunc extends RichSinkFunction<String> {

    Connection conn;

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        Class.forName("org.apache.phoenix.jdbc.PhoenixDriver");
        conn= DriverManager.getConnection("jdbc:phoenix:hadoop-single:2181");
    }

    @Override
    public void close() throws Exception {
        super.close();
        if(conn!=null)
            conn.close();
    }

    @Override
    public void invoke(String value, Context context) throws Exception {
        JSONObject jsonObject= JSON.parseObject(value);

        String table=jsonObject.getString("table");
        String id=jsonObject.getJSONObject("data").getString("id");
        String data=jsonObject.getJSONObject("data").toJSONString();

        String sql="upsert into dim8_"+table+ " (id,line) values (?,?) ";

        PreparedStatement ps=null;
        try
        {
            ps= conn.prepareStatement(sql);
            ps.setString(1,id);
            ps.setString(2,data);
            ps.execute();
            conn.commit();
        }catch (Exception e)
        {
            e.printStackTrace();
        }finally {
            if(ps!=null)
                ps.close();
        }

    }
}
